package com.batch.quartz.config;

import com.batch.quartz.batch.listener.JobCompletionNotificationListener;
import com.batch.quartz.entity.Person;
import com.batch.quartz.job.ImportUserQuartzJob;
import com.batch.quartz.reader.PersonItemProcessor;
import org.quartz.JobDataMap;
import org.quartz.Scheduler;
import org.quartz.SchedulerContext;
import org.quartz.spi.JobFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.JobLocator;
import org.springframework.batch.core.configuration.JobRegistry;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.support.JobRegistryBeanPostProcessor;
import org.springframework.batch.core.configuration.support.MapJobRegistry;
import org.springframework.batch.core.explore.support.JobExplorerFactoryBean;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.quartz.CronTriggerFactoryBean;
import org.springframework.scheduling.quartz.JobDetailFactoryBean;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;

import javax.sql.DataSource;
import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.Executor;

/**
 * <h3>partitionjob</h3>
 * <p>quartz 配置</p>
 *
 * @author : hduong
 * @version : 1.0
 * @date : 2019-12-20 11:25
 **/
@Configuration
public class QuartzConfig extends CommonConfig {
    public QuartzConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        super(jobBuilderFactory, stepBuilderFactory);
    }
    //1: quartz.properties配置
    //2: 任务bean jobDetail配置
    //3： 调度器 schedulerBean配置
    //4: 注册 启动配置
    //5： 线程池配置
    //6： autowiringJobFactory() 配置

    @Autowired
    private JobLauncher jobLauncher;

    @Autowired
    private JobLocator jobLocator;

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(JobRegistry jobRegistry) {
        JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor = new JobRegistryBeanPostProcessor();
        jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry);
        return jobRegistryBeanPostProcessor;
    }

//    @Bean
//    public SchedulerFactoryBean schedulerFactoryBean() throws IOException {
//        SchedulerFactoryBean factory = new SchedulerFactoryBean();
//
//        factory.setSchedulerName("Cluster_Scheduler");
//        factory.setDataSource(dataSource);
//        factory.setApplicationContextSchedulerContextKey("applicationContext");
//        //设置任务执行器
//        factory.setTaskExecutor(schedulerThreadPool());
//        //配置这个 支持 序列化bean
//        SchedulerContext schedulerContext = new SchedulerContext();
//        schedulerContext.put("jobName", "importUserJob");
//        schedulerContext.put("jobLauncher", jobLauncher);
//        schedulerContext.put("jobLocator", jobLocator);
//        factory.setSchedulerContextAsMap(schedulerContext);
//
//        //设置触发器 : 绑定了 jobDetail 任务定义
//        factory.setTriggers(quartzTrigger().getObject());
//        //设置quartz属性
//        factory.setQuartzProperties(quartzProperties());
//        //job工厂
//        factory.setJobFactory(autowiringJobFactory());
//        factory.setOverwriteExistingJobs(true);
//        factory.setStartupDelay(15);
//        return factory;
//    }


    @Bean
    public SchedulerFactoryBean schedulerFactoryBean() throws IOException {
        SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
        schedulerFactoryBean.setTriggers(quartzTrigger().getObject());
        return schedulerFactoryBean;
    }

    @Bean
    public CronTriggerFactoryBean quartzTrigger() {
        CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();

        cronTriggerFactoryBean.setJobDetail(importUserJob1().getObject());
        cronTriggerFactoryBean.setCronExpression("0/15 * * * * ?");
        return cronTriggerFactoryBean;
    }

    /**
     * 启动任务配置 不能少
     *
     * @param
     * @return : org.quartz.Scheduler
     * @date : 2019/12/20 14:43
     * @author : hduong
     */
    @Bean
    public Scheduler scheduler() throws Exception {
        Scheduler scheduler = schedulerFactoryBean().getScheduler();
        scheduler.start();
        return scheduler;
    }


    @Bean
    public JobDetailFactoryBean importUserJob1() {
        JobDetailFactoryBean jobDetailFactoryBean = new JobDetailFactoryBean();

        jobDetailFactoryBean.setJobClass(ImportUserQuartzJob.class);
        jobDetailFactoryBean.setDurability(true);
        jobDetailFactoryBean.setGroup("Import_User_Group");
        jobDetailFactoryBean.setName("Import_User_Name");
        //使用 schedulerContext 可以序列化:
//        SchedulerContext schedulerContext = new SchedulerContext();
//        schedulerContext.put("jobName", "importUserJob");
//        schedulerContext.put("jobLauncher", jobLauncher);
//        schedulerContext.put("jobLocator", jobRepository);
//        jobDetailFactoryBean.setJobDataAsMap(schedulerContext);
//        jobDetailFactoryBean.set
        //jobDetailFactoryBean.setJobDataMap(jobDataMap);//配置这个无法序列化 locator
        //jobDetailFactoryBean.setJobDataAsMap(jobDataMap);
        jobDetailFactoryBean.setRequestsRecovery(true);
        return jobDetailFactoryBean;
    }

    /**
     * 未配置启动的任务 : 可以手动启动 TODO
     *
     * @param
     * @return : org.springframework.scheduling.quartz.JobDetailFactoryBean
     * @date : 2019/12/20 15:04
     * @author : hduong
     */
//    @Bean
//    public JobDetailFactoryBean quartzJob() {
//        JobDetailFactoryBean jobDetailFactoryBean = new JobDetailFactoryBean();
//
//        jobDetailFactoryBean.setJobClass(QuartzJob.class);
//        jobDetailFactoryBean.setDurability(true);
//        jobDetailFactoryBean.setGroup("piLiang");
//        jobDetailFactoryBean.setName("piLiang_job");
//        jobDetailFactoryBean.setRequestsRecovery(true);
//        return jobDetailFactoryBean;
//    }
    @Bean
    public Properties quartzProperties() throws IOException {
        PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
        propertiesFactoryBean.setLocation(new ClassPathResource("/quartz.properties"));

        //在quartz.properties中的属性被读取并注入后再初始化对象
        propertiesFactoryBean.afterPropertiesSet();
        return propertiesFactoryBean.getObject();
    }


    @Bean
    public JobFactory autowiringJobFactory() {
        return new AutowiringJobFactory();
    }

    @Bean
    public Executor schedulerThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();

        executor.setCorePoolSize(15);
        executor.setMaxPoolSize(25);
        executor.setQueueCapacity(100);
        return executor;
    }

    @Autowired
    private JobCompletionNotificationListener listener;

    @Bean(name = "importUserJob")
    public Job importUserJob() {
        return jobBuilderFactory.get("importUserJob")
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .flow(step1())
                .end()
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<Person, Person>chunk(10)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }


//    @Bean
//    @StepScope
//    public CommonFileReader reader(@Value("#{jobParameters[fileName]}") String fileName) {
//        return new CommonFileReader(fileName, Person.class, ",",null);
//    }

    @Bean
    public FlatFileItemReader<Person> reader() {
        return new FlatFileItemReaderBuilder<Person>()
                .name("personItemReader")
                .resource(new ClassPathResource("person.txt"))
                .delimited()
                .names(new String[]{"firstName", "lastName"})
                .fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                    setTargetType(Person.class);
                }})
                .build();
    }

    @Bean
    public PersonItemProcessor processor() {
        return new PersonItemProcessor();
    }

    @Autowired
    DataSource dataSource;

    @Bean
    public JdbcBatchItemWriter<Person> writer() {
        return new JdbcBatchItemWriterBuilder<Person>()
                .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
                .sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
                .dataSource(dataSource)
                .build();
    }

    //********************************** 20200901 新增内容 *与quarter整合调度
    /*
    注册job，把job注入到容器中，在jobLauncher启动job时才能从容器中获取要启动的Job
    任务浏览器（JObExplorer）”是“任务存储器（JobRepository）”的只读版本，
                像后者一样，它可以通过工厂bean简单的配置生成
     */
    @Bean
    public MapJobRegistry jobRegistry() {
        return new MapJobRegistry();
    }

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() {
        JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor = new JobRegistryBeanPostProcessor();
        jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry());
        return jobRegistryBeanPostProcessor;
    }

    @Bean
    public JobExplorerFactoryBean jobExplorerFactoryBean() {
        JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
        jobExplorerFactoryBean.setDataSource(dataSource);
        return jobExplorerFactoryBean;
    }

    /*
    Quartz定时调度配置
    要执行的类,继承QuartzJobBean,在executeInternal方法启动job
     */
    @Bean
    public JobDetailFactoryBean jobDetailFactoryBean() {
        JobDetailFactoryBean jobDetailFactoryBean = new JobDetailFactoryBean();
        JobDataMap jobDataMap = new JobDataMap();
        jobDataMap.put("job", importUserJob());
        jobDataMap.put("jobLauncher", jobLauncher);
        jobDetailFactoryBean.setJobDataMap(jobDataMap);
        jobDetailFactoryBean.setJobClass(QuartzJobLauncher.class);
        return jobDetailFactoryBean;
    }






}
